﻿#region 从另外一个topic收数据 
using Confluent.Kafka;
using CSRedis;
using Google.Protobuf;
using Newtonsoft.Json;
using QskjLibrary;
using QskjServiceKafka.Dto;
using QskjServiceKafka.Service;
using System;
using System.Data;
using static Org.BouncyCastle.Bcpg.Attr.ImageAttrib;
using System.Globalization;
using System.Security.Cryptography;
using System.Text;
#region 消费数据
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("DEVICE_STATUS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始1");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_STATUS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始2");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_GPS");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始3");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_ALARM");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始4");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("VEHICLE_ALARM_FILE");
    Console.WriteLine(DateTime.Now.ToString() + "qskj设备状态GPS告警数据kafka消费开始5");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
Task.Factory.StartNew(() =>
{
    string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    List<string> topics = new List<string>();
    topics.Add("topic-media-complete");
    Console.WriteLine(DateTime.Now.ToString() + "qskj直连设备报警数据接收开始6");

    //Thread.Sleep(1000 * 20);
    try
    {
        Consumer(brokerList, topics);
    }
    catch (Exception e)
    {
        Console.WriteLine("qskj直连设备报警数据接收-数据消费异常" + e);
    }
    while (true)
    {
        Thread.Sleep(1000 * 200);
    }
});
#endregion
#endregion
while (true)
{
    Thread.Sleep(5000);
}
async void PostMsgToDingDing(string msg)
{
    //var client = new HttpClient();
    //var request = new HttpRequestMessage(HttpMethod.Post, "https://oapi.dingtalk.com/robot/send?access_token=1bdf2e7e2a2dd676ace21cb146b2b6e5c7a34fd39eae8eb9d74a5e52096f3a1d");
    //var content = new StringContent("{\r\n    \"msgtype\":\"text\",\r\n    \"text\":\r\n    {\r\n        \"content\":\"爱丽丝端口服务666- " + msg + "\"\r\n    }\r\n}", null, "application/json");
    //request.Content = content;
    //var response = await client.SendAsync(request);
    //response.EnsureSuccessStatusCode(); 
}
void DeviceStatusDataToMysql(DeviceStatus status, CSRedisClient redisDb)
{
    try
    {
        string zl = redisDb.Get($@"zl_{status.devGbId}");
        if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
        {
            return;
        }
        DateTimeOffset dateTimeOffset = DateTimeOffset.FromUnixTimeMilliseconds(Convert.ToInt64(status.timestamp));
        DateTime dt = dateTimeOffset.LocalDateTime;
        string dtStr = dt.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Device_Status{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Device_Status{dtStr}
                        (
                           id                   bigint not null auto_increment,
                           vehicleNo            varchar(50) comment '车牌号',
                           devGbId              varchar(50) comment '设备国标id',
                           devStatus            varchar(10) comment '设备在线状态1在线online0离线offline',
                           timestamp            bigint comment '上报时间戳',
                           accessPlatformNode   varchar(50) comment '设备接入的标识，用于区分设备状态是哪一个设备接入上报',
                           createtime           varchar(50),
                           platform             varchar(50),
                           createtimeSql           datetime,
                           INDEX index_vehicleNo (vehicleNo),
                           INDEX index_devStatus (devStatus),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string sql = $@"insert into qskjrecord.B_Device_Status{dtStr}(vehicleNo,devGbId,devStatus,timestamp,accessPlatformNode,createtime,platform,createtimeSql)
                        values('{status.devGbId}','{status.devGbId}','{status.devStatus}','{status.timestamp}','{status.accessPlatformNode}','{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{status.platform}',now())";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);
    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka设备状态信息写入数据库异常");
        Console.WriteLine("设备状态信息写入数据库异常" + JsonConvert.SerializeObject(e));
    }
}
void VehicleGpsDataToMysql(VehicleGps status, CSRedisClient redisDb)
{
    string dtL = string.Empty;
    try
    {
        string zl = redisDb.Get($@"zl_{status.devGbId}");
        if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
        {
            return;
        }
        DateTimeOffset dateTimeOffset = DateTimeOffset.FromUnixTimeMilliseconds(Convert.ToInt64(status.timestamp));
        DateTime dt = dateTimeOffset.LocalDateTime;
        string dtStr = dt.ToString("yyyyMMdd"); dtL = dtStr;
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_GPS{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_GPS{dtStr}
                        (
                           id                   bigint not null auto_increment,
                           vehicleNo            varchar(50) comment '车牌号',
                           devGbId              varchar(50) comment '设备国标id',
                           latitude             varchar(50) comment '经度 （必填）格式ddd.ddddddddd (默认WGS84坐标)',
                           longitude            varchar(50) comment '纬度（必填）格式
                                     ddd.ddddddddd (默认WGS84坐标)',
                           direction            varchar(50) comment '方向（非必填） 取值为当前摄像头方向与正北方的顺时针夹角,取值范围 0°~360° ,单位:(° )',
                           speed                varchar(50) comment '速度（非必填） 速度（Km/s）',
                           altitude             varchar(50) comment '海拔高度（非必填） 高度(m)',
                           timestamp            varchar(50) comment '上报时间戳',
                           createtime           datetime,
                           carState             varchar(50),
                           platform             varchar(50),
                           createtimeSql           datetime,
                           INDEX index_vehicleNo (vehicleNo),
                           INDEX index_createtime (createtime),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_GPS{dtStr}(vehicleNo,devGbId,latitude,longitude,direction,speed,altitude,timestamp,createtime,carState,platform,createtimeSql)
                        values('{status.devGbId}','{status.devGbId}','{status.latitude}','{status.longitude}','{status.direction}','{status.speed}','{status.altitude}','{status.timestamp}','{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{status.carState}','{status.platform}',now())";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);

    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka设备GPS信息写入数据库异常");
        Console.WriteLine(dtL + "设备GPS信息写入数据库异常" + JsonConvert.SerializeObject(e));
    }
}
void VehicleAlarmDataToMysql(VehicleAlarm status, CSRedisClient redisDb5, CSRedisClient redisDb)
{
    try
    {
        string zl = redisDb.Get($@"zl_{status.devGbId}");
        if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
        {
            return;
        }
        DateTimeOffset dateTimeOffset = DateTimeOffset.FromUnixTimeMilliseconds(Convert.ToInt64(status.timestamp));
        DateTime dt = dateTimeOffset.LocalDateTime;
        string dtStr = dt.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_Alarm{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_Alarm{dtStr}
                        (
                           id                   bigint not null auto_increment,
                           devGbId              varchar(50) comment '设备国标id',
                           alarmId              varchar(50) comment '报警id',
                           vehicleNo            varchar(50) comment '车牌号',
                           alarmMainType        int comment '1主动安全DSM报警2主动安全ADAS报警3行业监管报警4设备报警',
                           alarmSubType         int comment '1.1抽烟1.2接打电话1.3疲劳驾驶1.4分神1.5驾驶员异常1.6摄像头遮挡1.7红外阻隔型墨镜报警2.1前向碰撞预警2.2车道偏离2.3车距过近3.1超速3.2连续行驶疲劳驾驶3.3两到五点运营4.1GNSS模块发生故障4.2GNSS天线未接或被剪断4.3GNSS天线短路4.4终端主电源欠压4.5终端主电源掉电4.6摄像头故障',
                           alarmType            varchar(50) comment '1.1抽烟1.2接打电话1.3疲劳驾驶1.4分神1.5驾驶员异常1.6摄像头遮挡1.7红外阻隔型墨镜报警2.1前向碰撞预警2.2车道偏离2.3车距过近3.1超速3.2连续行驶疲劳驾驶3.3两到五点运营4.1GNSS模块发生故障4.2GNSS天线未接或被剪断4.3GNSS天线短路4.4终端主电源欠压4.5终端主电源掉电4.6摄像头故障',
                           alarmTime            datetime comment '告警时间yyyy-MM-dd HH:mm:ss',
                           latitude             varchar(50) comment '经度 （必填）格式ddd.ddddddddd (默认WGS84坐标)',
                           longitude            varchar(50) comment '纬度（必填）格式
                                     ddd.ddddddddd (默认WGS84坐标)',
                           speed                varchar(50) comment '速度（非必填） 速度（Km/s）',
                           url                  varchar(300) comment '报警附件',
                           platform             varchar(50) comment '车载平台厂商1中安2锐明3海康',
                           timestamp            bigint comment '上报时间戳',
                           createtime           datetime,
                           driverId             varchar(50),
                           driverWId            varchar(50),
                           driverName           varchar(50), 
                           createtimeSql           datetime,
                           INDEX index_vehicleNo (vehicleNo),
                           INDEX index_createtime (createtime),
                           INDEX index_driverName (driverName),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string drStr = redisDb5.Get(status.vehicleNo.Substring(1));//出租车不带沪
        string drId = string.Empty, drWId = string.Empty, drNa = string.Empty;
        if (!string.IsNullOrEmpty(drStr))
        {
            string[] drArr = drStr.Split(';');
            drId = drArr[0]; drWId = drArr[1]; drNa = drArr[2];
        }
        else
        {
            drStr = redisDb5.Get(status.vehicleNo);//大巴车带沪
            if (!string.IsNullOrEmpty(drStr))
            {
                string[] drArr = drStr.Split(';');
                drId = drArr[0]; drWId = drArr[1]; drNa = drArr[2];
            }
        }
        int first = 0; string group = string.Empty;
        string alarmPilao = redisDb.Get(status.vehicleNo + "3");//疲劳
        string alarmChaosu = redisDb.Get(status.vehicleNo + "9");//超速
        string alarmChouyan = redisDb.Get(status.vehicleNo + "1");//抽烟
        if (status.platform == "RUIMING" && (status.alarmType == "35" || status.alarmType == "87" || status.alarmType == "113"))//疲劳
        {
            if (string.IsNullOrEmpty(alarmPilao))
            {
                redisDb.Set(status.vehicleNo + "3", status.alarmId, new TimeSpan(0, 1, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmPilao;
            }
        }
        else if (status.platform == "RUIMING" && (status.alarmType == "17" || status.alarmType == "34" || status.alarmType == "44"))//超速
        {
            if (string.IsNullOrEmpty(alarmChaosu))
            {
                redisDb.Set(status.vehicleNo + "9", status.alarmId, new TimeSpan(0, 5, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmChaosu;
            }
        }
        else if (status.platform == "RUIMING" && (status.alarmType == "89"))//抽烟
        {
            if (string.IsNullOrEmpty(alarmChouyan))
            {
                redisDb.Set(status.vehicleNo + "1", status.alarmId, new TimeSpan(0, 5, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmChouyan;
            }
        }
        else if (status.platform == "ZAKJ" && (status.alarmType == "5" || status.alarmType == "43"))//疲劳
        {
            if (string.IsNullOrEmpty(alarmPilao))
            {
                redisDb.Set(status.vehicleNo + "3", status.alarmId, new TimeSpan(0, 1, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmPilao;
            }
        }
        else if (status.platform == "ZAKJ" && (status.alarmType == "39"))//超速
        {
            if (string.IsNullOrEmpty(alarmChaosu))
            {
                redisDb.Set(status.vehicleNo + "9", status.alarmId, new TimeSpan(0, 5, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmChaosu;
            }
        }
        else if (status.platform == "ZAKJ" && (status.alarmType == "1"))//抽烟
        {
            if (string.IsNullOrEmpty(alarmChouyan))
            {
                redisDb.Set(status.vehicleNo + "1", status.alarmId, new TimeSpan(0, 5, 0));
                first = 1; group = status.alarmId;
            }
            else
            {
                first = 0; group = alarmChouyan;
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_Alarm{dtStr}(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);


        string sqlYear = $@"insert into qskjrecord.B_Vehicle_Alarm (devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
        int resYear = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYear);
        if ((status.platform == "RUIMING") && (status.alarmType == "1" || status.alarmType == "2"))
        {
            string sqlYjbj = $@"insert into qskjrecord.B_Vehicle_AlarmYjbj(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
            int resYjbj = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYjbj);
        }
        if (status.platform == "zl" && status.alarmType == "6574")
        {
            string sqlYjbj = $@"insert into qskjrecord.B_Vehicle_AlarmYjbj(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
            int resYjbj = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYjbj);
        }
        if ((status.platform == "zl" && status.alarmType == "6506") || (status.platform == "ZAKJ" && (status.alarmType == "14" || status.alarmType == "135")) || (status.platform == "RUIMING" && (status.alarmType == "10" || status.alarmType == "99")))
        {
            string sqlYjbj = $@"insert into qskjrecord.B_Vehicle_AlarmZdgz(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{dt.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
            int resYjbj = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYjbj);
        }
    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka设备报警信息写入数据库异常");
        Console.WriteLine("设备报警信息写入数据库异常" + JsonConvert.SerializeObject(e));
    }
}
void VehicleAlarmFILEDataToMysql(VehicleAlarm status, CSRedisClient redisDb)
{
    try
    {
        string zl = redisDb.Get($@"zl_{status.devGbId}");
        if (!string.IsNullOrWhiteSpace(zl) && (status.platform == "RUIMING" || status.platform == "ZAKJ"))
        {
            return;
        }
        string dtStr = DateTime.Now.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_AlarmFile{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_AlarmFile{dtStr}
                        (
                           id                   bigint not null auto_increment, 
                           alarmId              varchar(300) comment '报警id', 
                           url                  text comment '报警附件',
                           createtime           datetime,
                           IsInpector           int default 0,
                           INDEX index_createtime (createtime),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_AlarmFile{dtStr}(alarmId,url,createtime)
                        values('{status.alarmId}','{status.url}','{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}')";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);

    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka设备报警附件信息更新写入数据库异常");
        Console.WriteLine("设备报警附件信息更新写入数据库异常" + JsonConvert.SerializeObject(e));
    }
}
///直连设备报警数据写入数据库
int VehicleAlarmZlToMysql(VehicleAlarmZl status)
{
    try
    {
        string dtStr = DateTime.Now.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_AlarmZl{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_AlarmZl{dtStr}
                        (
                           id                   bigint not null auto_increment,
                           FileLength           int,
                           RelativePath         varchar(300),
                           AlarmID              varchar(300),
                           AttachLength         int,
                           AttachPath           text,
                           Status               int,
                           createtime           datetime,
                           AlarmIDMd5           varchar(300),
                           latitude             varchar(30),
                           longitude            varchar(30),
                           speed                varchar(30),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_AlarmZl{dtStr}(FileLength,RelativePath,AlarmID,AttachLength,AttachPath,Status,createtime,AlarmIDMd5,longitude,latitude,speed)
                        values({status.FileLength},'{status.RelativePath}','{status.AlarmID}',{status.AttachLength},'{status.AttachPath}',{status.Status},'{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}','{status.AlarmIDMd5}','{status.longitude}','{status.latitude}','{status.speed}');SELECT LAST_INSERT_ID();";
        object ob = MySqlHelper.ExecuteScalar(System.Data.CommandType.Text, sql);
        int res = ob == DBNull.Value ? 0 : Convert.ToInt32(ob);
        return res;
    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka直连设备报警数据写入数据库异常");
        Console.WriteLine("直连设备报警数据写入数据库异常" + JsonConvert.SerializeObject(e));
        return -1;
    }
}
///直连设备报警数据写入报警文件表
void VehicleZlAlarmFILEDataToMysql(VehicleAlarm status)
{
    try
    {
        string dtStr = DateTime.Now.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_AlarmFile{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_AlarmFile{dtStr}
                        (
                           id                   bigint not null auto_increment, 
                           alarmId              varchar(300) comment '报警id', 
                           url                  text comment '报警附件',
                           createtime           datetime,
                           IsInpector           int default 0,
                           INDEX index_createtime (createtime),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_AlarmFile{dtStr}(alarmId,url,createtime)
                        values('{status.alarmId}','{status.url}','{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}')";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);

    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka直连设备报警数据写入报警文件表异常");
        Console.WriteLine("直连设备报警数据写入报警文件表异常" + JsonConvert.SerializeObject(e));
    }
}
void VehicleZlAlarmDataToMysql(VehicleAlarm status, CSRedisClient redisDb5)
{
    try
    {
        string dtStr = DateTime.Now.ToString("yyyyMMdd");
        if (DateTime.Now.Hour < 7)
        {
            string exists = $@"SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA = 'qskjrecord' AND TABLE_NAME = 'B_Vehicle_Alarm{dtStr}';";
            object resExi = MySqlHelper.ExecuteScalar(CommandType.Text, exists);
            if (resExi == null || resExi == DBNull.Value)
            {
                string sqlCt = $@"create table IF NOT EXISTS qskjrecord.B_Vehicle_Alarm{dtStr}
                        (
                           id                   bigint not null auto_increment,
                           devGbId              varchar(50) comment '设备国标id',
                           alarmId              varchar(300) comment '报警id',
                           vehicleNo            varchar(50) comment '车牌号',
                           alarmMainType        int comment '1主动安全DSM报警2主动安全ADAS报警3行业监管报警4设备报警',
                           alarmSubType         int comment '1.1抽烟1.2接打电话1.3疲劳驾驶1.4分神1.5驾驶员异常1.6摄像头遮挡1.7红外阻隔型墨镜报警2.1前向碰撞预警2.2车道偏离2.3车距过近3.1超速3.2连续行驶疲劳驾驶3.3两到五点运营4.1GNSS模块发生故障4.2GNSS天线未接或被剪断4.3GNSS天线短路4.4终端主电源欠压4.5终端主电源掉电4.6摄像头故障',
                           alarmType            varchar(50) comment '1.1抽烟1.2接打电话1.3疲劳驾驶1.4分神1.5驾驶员异常1.6摄像头遮挡1.7红外阻隔型墨镜报警2.1前向碰撞预警2.2车道偏离2.3车距过近3.1超速3.2连续行驶疲劳驾驶3.3两到五点运营4.1GNSS模块发生故障4.2GNSS天线未接或被剪断4.3GNSS天线短路4.4终端主电源欠压4.5终端主电源掉电4.6摄像头故障',
                           alarmTime            datetime comment '告警时间yyyy-MM-dd HH:mm:ss',
                           latitude             varchar(50) comment '经度 （必填）格式ddd.ddddddddd (默认WGS84坐标)',
                           longitude            varchar(50) comment '纬度（必填）格式
                                     ddd.ddddddddd (默认WGS84坐标)',
                           speed                varchar(50) comment '速度（非必填） 速度（Km/s）',
                           url                  varchar(300) comment '报警附件',
                           platform             varchar(50) comment '车载平台厂商1中安2锐明3海康',
                           timestamp            bigint comment '上报时间戳',
                           createtime           datetime,
                           driverId           varchar(50),
                           driverWId           varchar(50),
                           driverName           varchar(50),
                           createtimeSql           datetime,
                           INDEX index_vehicleNo (vehicleNo),
                           INDEX index_createtime (createtime),
                           INDEX index_driverName (driverName),
                           primary key (id)
                        );";
                MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlCt);
            }
        }
        string drStr = redisDb5.Get(status.vehicleNo.Substring(1));//出租车不带沪
        string drId = string.Empty, drWId = string.Empty, drNa = string.Empty;
        if (!string.IsNullOrEmpty(drStr))
        {
            string[] drArr = drStr.Split(';');
            drId = drArr[0]; drWId = drArr[1]; drNa = drArr[2];
        }
        else
        {
            drStr = redisDb5.Get(status.vehicleNo);//大巴车带沪
            if (!string.IsNullOrEmpty(drStr))
            {
                string[] drArr = drStr.Split(';');
                drId = drArr[0]; drWId = drArr[1]; drNa = drArr[2];
            }
        }
        string sql = $@"insert into qskjrecord.B_Vehicle_Alarm{dtStr}(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
        int res = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sql);


        string sqlYear = $@"insert into qskjrecord.B_Vehicle_Alarm (devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
        int resYear = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYear);
        if ((status.platform == "RUIMING") && (status.alarmType == "1" || status.alarmType == "2"))
        {
            string sqlYjbj = $@"insert into qskjrecord.B_Vehicle_AlarmYjbj(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
            int resYjbj = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYjbj);
        }
        if (status.platform == "zl" && status.alarmType == "6574")
        {
            string sqlYjbj = $@"insert into qskjrecord.B_Vehicle_AlarmYjbj(devGbId,alarmId,vehicleNo,alarmMainType,alarmSubType,alarmType,alarmTime,latitude,longitude,speed,url,platform,timestamp,createtime,driverId,driverWId,driverName,createtimeSql)
                        values('{status.devGbId}','{status.alarmId}','{status.vehicleNo}',{status.alarmMainType},{status.alarmSubType},'{status.alarmType}','{status.alarmTime}','{status.latitude}','{status.longitude}','{status.speed}','{status.url}','{status.platform}',{status.timestamp},'{DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss")}','{drId}','{drWId}','{drNa}',now())";
            int resYjbj = MySqlHelper.ExecuteNonQuery(System.Data.CommandType.Text, sqlYjbj);
        }
    }
    catch (Exception e)
    {
        PostMsgToDingDing("QskjServiceKafka设备报警信息写入数据库异常");
        Console.WriteLine("设备报警信息写入数据库异常" + JsonConvert.SerializeObject(e));
    }
}

async void Producer(string brokerList, string topicName, string text)
{
    //Console.WriteLine(topicName + "数据生产");
    var config = new ProducerConfig { BootstrapServers = brokerList };

    using (var producer = new ProducerBuilder<string, string>(config).Build())
    {
        //Console.WriteLine($"Producer {producer.Name} producing on topic {topicName}.");

        {
            string key = Guid.NewGuid().ToString();
            string val = text;
            try
            {
                // Note: Awaiting the asynchronous produce request below prevents flow of execution
                // from proceeding until the acknowledgement from the broker is received (at the 
                // expense of low throughput).
                var deliveryReport = await producer.ProduceAsync(
                    topicName, new Message<string, string> { Key = key, Value = val });

                //Console.WriteLine($"delivered to: {deliveryReport.TopicPartitionOffset}");
                Console.WriteLine($"--vehno--");
            }
            catch (ProduceException<string, string> e)
            {
                Console.WriteLine($"failed to deliver message: {e.Message} [{e.Error.Code}] Topic:{topicName}---生产数据失败");
            }
        }

    }
}
async void Consumer(string brokerList, List<string> topics)
{
    CancellationTokenSource cts = new CancellationTokenSource();
    Console.CancelKeyPress += (_, e) =>
    {
        e.Cancel = true; // prevent the process from terminating.
        cts.Cancel();
    };
    Run_Consume(brokerList, topics, cts.Token);
}
/// <summary>
///     In this example
///         - offsets are automatically committed.
///         - no extra thread is created for the Poll (Consume) loop.
/// </summary>
void Run_Consume(string brokerList, List<string> topics, CancellationToken cancellationToken)
{
    CSRedisClient redisDb5 = null;
    if (redisDb5 == null)
    {
        redisDb5 = RedisHelp.GetInstance5();
    }
    CSRedisClient redisDb = null;
    if (redisDb == null)
    {
        redisDb = RedisHelp.GetInstance();
    }
    //string brokerList = "192.168.3.52:9092,192.168.3.146:9092,192.168.3.164:9092";
    var config = new ConsumerConfig
    {
        BootstrapServers = brokerList,
        GroupId = "DEVICE_VEHICLE_STATUS_VEHICLE_GPS_ALARM12",
        EnableAutoOffsetStore = false,
        EnableAutoCommit = true,
        StatisticsIntervalMs = 5000,
        SessionTimeoutMs = 6000,
        AutoOffsetReset = AutoOffsetReset.Latest,
        EnablePartitionEof = true,
        // A good introduction to the CooperativeSticky assignor and incremental rebalancing:
        // https://www.confluent.io/blog/cooperative-rebalancing-in-kafka-streams-consumer-ksqldb/
        PartitionAssignmentStrategy = PartitionAssignmentStrategy.CooperativeSticky
    };

    // Note: If a key or value deserializer is not set (as is the case below), the 
    // deserializer corresponding to the appropriate type from Confluent.Kafka.Deserializers
    // will be used automatically (where available). The default deserializer for string
    // is UTF8. The default deserializer for Ignore returns null for all input data
    // (including non-null data).
    using (var consumer = new ConsumerBuilder<Ignore, string>(config)
        // Note: All handlers are called on the main .Consume thread.
        .SetErrorHandler((_, e) => Console.WriteLine($"Error: {e.Reason}"))
        .SetStatisticsHandler((_, json) =>
            //Console.WriteLine($"Statistics: {json}")
            Console.WriteLine("rw")
            )
        .SetPartitionsAssignedHandler((c, partitions) =>
        {
            // Since a cooperative assignor (CooperativeSticky) has been configured, the
            // partition assignment is incremental (adds partitions to any existing assignment).
            //Console.WriteLine(
            //    "Partitions incrementally assigned: [" +
            //    string.Join(',', partitions.Select(p => p.Partition.Value)) +
            //    "], all: [" +
            //    string.Join(',', c.Assignment.Concat(partitions).Select(p => p.Partition.Value)) +
            //    "]");

            // Possibly manually specify start offsets by returning a list of topic/partition/offsets
            // to assign to, e.g.:
            // return partitions.Select(tp => new TopicPartitionOffset(tp, externalOffsets[tp]));
        })
        .SetPartitionsRevokedHandler((c, partitions) =>
        {
            // Since a cooperative assignor (CooperativeSticky) has been configured, the revoked
            // assignment is incremental (may remove only some partitions of the current assignment).
            var remaining = c.Assignment.Where(atp => partitions.Where(rtp => rtp.TopicPartition == atp).Count() == 0);
            //Console.WriteLine(
            //    "Partitions incrementally revoked: [" +
            //    string.Join(',', partitions.Select(p => p.Partition.Value)) +
            //    "], remaining: [" +
            //    string.Join(',', remaining.Select(p => p.Partition.Value)) +
            //    "]");
        })
        .SetPartitionsLostHandler((c, partitions) =>
        {
            // The lost partitions handler is called when the consumer detects that it has lost ownership
            // of its assignment (fallen out of the group).
            //Console.WriteLine($"Partitions were lost: [{string.Join(", ", partitions)}]");
        })
        .Build())
    {
        consumer.Subscribe(topics);

        try
        {
            while (true)
            {
                try
                {
                    var consumeResult = consumer.Consume(cancellationToken);

                    if (consumeResult.IsPartitionEOF)
                    {
                        //Console.WriteLine(
                        //    $"Reached end of topic {consumeResult.Topic}, partition {consumeResult.Partition}, offset {consumeResult.Offset}.");

                        continue;
                    }

                    //Console.WriteLine($"Received message at {consumeResult.TopicPartitionOffset}: {consumeResult.Message.Value}");
                    if (consumeResult.Topic == "DEVICE_STATUS")
                    {
                        try
                        {
                            DeviceStatus device = new DeviceStatus(); device = JsonConvert.DeserializeObject<DeviceStatus>(consumeResult.Message.Value);
                            DeviceStatusDataToMysql(device, redisDb);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_STATUS")
                    {
                        try
                        {
                            DeviceStatus device = new DeviceStatus(); device = JsonConvert.DeserializeObject<DeviceStatus>(consumeResult.Message.Value);
                            DeviceStatusDataToMysql(device, redisDb);
                            //Console.WriteLine(consumeResult.Message.Value);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_GPS")
                    {
                        try
                        {
                            VehicleGps vehicle = new VehicleGps(); vehicle = JsonConvert.DeserializeObject<VehicleGps>(consumeResult.Message.Value);
                            VehicleGpsDataToMysql(vehicle, redisDb);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_ALARM")
                    {
                        try
                        {
                            VehicleAlarm vehicle = new VehicleAlarm(); vehicle = JsonConvert.DeserializeObject<VehicleAlarm>(consumeResult.Message.Value);
                            VehicleAlarmDataToMysql(vehicle, redisDb5, redisDb);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "VEHICLE_ALARM_FILE")
                    {
                        try
                        {
                            VehicleAlarm vehicle = new VehicleAlarm(); vehicle = JsonConvert.DeserializeObject<VehicleAlarm>(consumeResult.Message.Value);
                            VehicleAlarmFILEDataToMysql(vehicle, redisDb);
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("数据转换失败" + consumeResult.Message.Value + "-------" + e.Message);
                        }
                    }
                    else if (consumeResult.Topic == "topic-media-complete")
                    {
                        try
                        {
                            VehicleAlarmZl vehicleZl = new VehicleAlarmZl(); vehicleZl = JsonConvert.DeserializeObject<VehicleAlarmZl>(consumeResult.Message.Value);
                            VehicleAlarm vehicle = new VehicleAlarm();
                            {
                                byte[] result = Encoding.Default.GetBytes(vehicleZl.AlarmID);//报警号过长使用MD5摘要处理
                                MD5 md5 = new MD5CryptoServiceProvider();
                                byte[] output = md5.ComputeHash(result);
                                vehicle.alarmId = BitConverter.ToString(output).Replace("-", "");
                                //Console.WriteLine(vehicle.alarmId);
                                vehicleZl.AlarmIDMd5 = vehicle.alarmId;
                            }
                            vehicle.longitude = vehicleZl.longitude;
                            vehicle.latitude = vehicleZl.latitude;
                            vehicle.speed = vehicleZl.speed;
                            int alarmId = VehicleAlarmZlToMysql(vehicleZl);
                            vehicle.url = vehicleZl.AttachPath;//报警数据此处可能为空值***************** 
                            if (string.IsNullOrEmpty(vehicle.url) || string.IsNullOrEmpty(vehicleZl.RelativePath))
                            {
                                continue;
                            }
                            string[] str = vehicleZl.RelativePath.Split('/');//报警数据此处可能为空值*****************
                            //Console.WriteLine("zl----------------d");
                            string zlTerKey = redisDb.Get("zl_" + str[1]);
                            //Console.WriteLine("zl----------------e");
                            if (!string.IsNullOrEmpty(zlTerKey))
                            {
                                //Console.WriteLine("zl----------------f0");
                                vehicle.vehicleNo = zlTerKey;
                                //Console.WriteLine("zl----------------f00");
                            }
                            else
                            {
                                //Console.WriteLine("zl----------------f1");
                                string sql = $@"select license_plate from qskj.B_Taxi where  zlterminalkey='{str[1]}'";
                                object o = MySqlHelper.ExecuteScalar(CommandType.Text, sql);
                                //Console.WriteLine("zl----------------f2");
                                vehicle.vehicleNo = o == null || o == DBNull.Value ? "xxx" : o.ToString();
                                //Console.WriteLine("zl----------------f3");
                                if (!string.IsNullOrEmpty(vehicle.vehicleNo))
                                {
                                    redisDb.Set("zl_" + str[1], vehicle?.vehicleNo);
                                    redisDb.Set("zl_" + vehicle?.vehicleNo, str[1]);
                                }
                                //Console.WriteLine("zl----------------f4");
                            }
                            if (vehicle.vehicleNo != "xxx")
                            {
                                {
                                    vehicleZl.VehicleNo = vehicle.vehicleNo;//车牌号
                                                                            //写入topic-media-complete-vehicleno
                                    Producer(brokerList, "topic-media-complete-vehicleno", JsonConvert.SerializeObject(vehicleZl));//车载设备告警数据
                                }
                                //Console.WriteLine("zl----------------g" + vehicle.alarmId);
                                string[] alarmInfo = str[2].Split('_');
                                string dateString = alarmInfo[4];// "20240620173040";
                                string format = "yyyyMMddHHmmss";
                                DateTime dateTime = DateTime.ParseExact(dateString, format, CultureInfo.InvariantCulture);
                                vehicle.alarmTime = dateTime.ToString("yyyy-MM-dd HH:mm:ss");
                                vehicle.alarmType = Convert.ToString(alarmInfo[2]);
                                vehicle.devGbId = vehicle?.vehicleNo;
                                vehicle.platform = "zl";

                                //Console.WriteLine("zl----------------1111111111111111111111111111111111111111111111111111");
                                VehicleZlAlarmFILEDataToMysql(vehicle);
                                //Console.WriteLine("zl----------------2222222222222222222222222222222222222222222222222222");
                                VehicleZlAlarmDataToMysql(vehicle, redisDb5);
                                //Console.WriteLine("zl----------------3333333333333333333333333333333333333333333333333333");
                            }
                        }
                        catch (Exception e)
                        {
                            Console.WriteLine("topic-media-complete数据处理失败" + consumeResult.Message.Value + "-------" + e.Message);
                            Console.WriteLine("异常信息：" + JsonConvert.SerializeObject(e));
                        }
                    }
                    else
                    {

                    }
                    try
                    {
                        // Store the offset associated with consumeResult to a local cache. Stored offsets are committed to Kafka by a background thread every AutoCommitIntervalMs. 
                        // The offset stored is actually the offset of the consumeResult + 1 since by convention, committed offsets specify the next message to consume. 
                        // If EnableAutoOffsetStore had been set to the default value true, the .NET client would automatically store offsets immediately prior to delivering messages to the application. 
                        // Explicitly storing offsets after processing gives at-least once semantics, the default behavior does not.
                        consumer.StoreOffset(consumeResult);
                    }
                    catch (KafkaException e)
                    {
                        Console.WriteLine($"Store Offset error: {e.Error.Reason}");
                    }
                }
                catch (ConsumeException e)
                {
                    Console.WriteLine($"Consume error: {e.Error.Reason}");
                }
            }
        }
        catch (OperationCanceledException)
        {
            Console.WriteLine("Closing consumer.");
            consumer.Close();
        }
    }
}
